/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.listeners;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import br.pucrio.inf.lac.konoha.hyuga.core.DDSLink;
import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellClusterEvent2;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellContentEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.GridClusterEvent;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * DG2CEP's final output listener. It receives a {@link GridClusterEvent} that represents the detected cluster.
 * 
 * This listener is triggered by the CEP engine when a {@link GridClusterEvent} is detected, that is, when a cluster is
 * formed and detected by DG2CEP. The listener will output the detection event to the interested parts, e.g., monitoring
 * applications, databases, benchmark servers, etc.
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class ClusterFormationListener {

	/** DDSLink (Data Stream). */
	private DDSLink ddsLink;

	// JSON Mapper
	ObjectMapper mapper;
	
	/** Thread Pool. */
	private ExecutorService	pool;
	
	/** LOGGER. */
	private static Logger LOGGER = Logger.getLogger(ClusterFormationListener.class.getCanonicalName());

	/**
	 * Default Constructor. Receive the DDSLink to send updates to management apps.
	 * @param ddsLink - DDSLink
	 */
	public ClusterFormationListener(DDSLink ddsLink) {
		this.ddsLink = ddsLink;
		this.mapper  = new ObjectMapper();
		this.pool    = Executors.newFixedThreadPool(16); 
	}
	
	public void update(GridClusterEvent gridCluster, Collection<?> borderNodes) {
	    System.out.println("aki");
	    int i = 0;
	    i++;
	}

	public void update(CellClusterEvent2[] coreCells, CellContentEvent[] borderNodes) {
		pool.submit(new ProcessCluster(new GridClusterEvent(coreCells, borderNodes)));
	}

	private class ProcessCluster implements Runnable {
	    GridClusterEvent cluster;
		
		public ProcessCluster(GridClusterEvent cluster) {
			this.cluster = cluster;
		}
		
		@Override
		public void run() {
			LOGGER.info(System.currentTimeMillis() + " DG2CEP DETECTED CLUSTER of SIZE: " + cluster.toString());

			Map<String, Object> clusterMap = new HashMap<String, Object>();
			clusterMap.put("operation", "add_cluster");

			CellClusterEvent2[] clusterCells = cluster.getCells();
			double[] x = new double[clusterCells.length]; // latitude index
			double[] y = new double[clusterCells.length]; // longitude index
			double[] s = new double[clusterCells.length]; // number of nodes (size)
			int totalNumNodes = 0;

			// Core Cells
			for (int i = 0; i < clusterCells.length; i++) {
				x[i] = clusterCells[i].getX();
				y[i] = clusterCells[i].getY();
				s[i] = clusterCells[i].numNodes();
				totalNumNodes += clusterCells[i].numNodes();
			}
			
	        // JSON Mapping
            clusterMap.put("latindex",   x);
            clusterMap.put("lngindex",   y);
            clusterMap.put("sizeindex",  s);
            clusterMap.put("core",       totalNumNodes);
            
			
			// Border Cells
			if (cluster.getNeighbors().length != 0) {
			    CellContentEvent[] neighborCells = cluster.getNeighbors();
			    double[] bx = new double[neighborCells.length]; // border latitude index
	            double[] by = new double[neighborCells.length]; // border longitude index
	            double[] bs = new double[neighborCells.length]; // border number of nodes (size)
	            
	            for (int i = 0; i < neighborCells.length; i++) {
	                bx[i] = neighborCells[i].getX();
	                by[i] = neighborCells[i].getY();
//	                bs[i] = neighborCells[i].numNodes();
//	                totalNumNodes += neighborCells[i].numNodes();
	            }
	            
	            Map<String, Object> neighborMap = new HashMap<String, Object>();
	            neighborMap.put("latindex",  bx);
	            neighborMap.put("lngindex",  by);
	            neighborMap.put("sizeindex", bs);
	            
	            clusterMap.put("neighbors", neighborMap);
			} else {
			    clusterMap.put("neighbors", null);
			}
			
            clusterMap.put("total", totalNumNodes);

            try {
				ddsLink.publishOut(mapper.writeValueAsBytes(clusterMap));
			} catch (JsonProcessingException e) {
				e.printStackTrace();
			}
		}
	}
}
